package com.bw.gmall.realtime.app.dwd;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import java.time.Duration;
/**
 * 交易域订单明细事实表应用
 * 负责处理订单明细数据，将订单数据转化为结构化的明细事实表
 * 主要功能点：
 * - 从Kafka读取订单相关数据
 * - 解析订单详情信息
 * - 关联商品、用户、促销等维度数据
 * - 提取订单金额、数量、状态等交易信息
 * - 将处理后的数据写入Kafka主题
 * 
 * 数据流向：
 * 订单相关数据 -> DwdTradeOrderDetail -> 处理后写入dwd_trade_order_detail主题
 */
public class DwdTradeOrderDetail {

    /**
     * 主方法，负责处理订单明细数据的ETL流程
     * 1. 获取Flink执行环境
     * 2. 从Kafka读取预处理订单数据
     * 3. 过滤出下单数据(insert类型)
     * 4. 创建DWD层订单明细表
     * 5. 将处理后的数据写入Kafka
     * 6. 启动任务执行
     *
     * @param args 命令行参数
     * @throws Exception 可能抛出的异常
     */
    public static void main(String[] args) throws Exception {

        //TODO 1.获取执行环境
        // 创建流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度，生产环境建议设置为Kafka主题分区数
        env.setParallelism(1);
        // 创建表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //1.3 设置状态的TTL  生产环境设置为最大乱序程度
        // 配置状态保留时间，防止状态无限增长
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5l));

        //TODO 2.创建topic_db表，直接从源数据库日志读取订单数据
        System.out.println("创建topic_db表...");
        tableEnv.executeSql(MyKafkaUtil.getTopicDb("dwd_trade_order_detail"));

        // 打印原始数据库日志，用于调试
        tableEnv.sqlQuery("select * from topic_db limit 10").execute().print();

        // 过滤出订单明细表和订单表数据
        System.out.println("过滤订单相关数据...");
        Table orderDataTable = tableEnv.sqlQuery("" +
                "select " +
                "    data['id'] id, " +
                "    data['order_id'] order_id, " +
                "    data['sku_id'] sku_id, " +
                "    data['sku_name'] sku_name, " +
                "    data['order_price'] order_price, " +
                "    data['sku_num'] sku_num, " +
                "    data['create_time'] create_time, " +
                "    data['source_type'] source_type_id, " +
                "    data['source_id'] source_id, " +
                "    data['split_total_amount'] split_total_amount, " +
                "    data['split_activity_amount'] split_activity_amount, " +
                "    data['split_coupon_amount'] split_coupon_amount, " +
                "    data['order_detail_activity_id'] activity_id, " +
                "    data['activity_rule_id'] activity_rule_id, " +
                "    data['order_detail_coupon_id'] order_detail_coupon_id, " +
                "    `type` " +
                "from topic_db " +
                "where `database` = 'gmall' " +
                "and `table` = 'order_detail' " +
                "and `type` = 'insert'");

        // 创建临时表
        tableEnv.createTemporaryView("dwd_order_pre", orderDataTable);
        tableEnv.toAppendStream(orderDataTable, Row.class).print("订单明细数据:");

        //TODO 3.简化查询，从预处理表获取所需字段
        System.out.println("构建最终订单明细数据...");
        Table filteredTable = tableEnv.sqlQuery("" +
                "select " +
                "id, " +  // 订单明细ID
                "order_id, " +  // 订单ID
                "'' as user_id, " +  // 用户ID（这里暂时使用空字符串，实际环境可能需要关联订单表）
                "sku_id, " +  // 商品SKU ID
                "sku_name, " +  // 商品名称
                "sku_num, " +  // 商品数量
                "order_price, " +  // 订单价格
                "'' as province_id, " +  // 省份ID（暂时使用空字符串）
                "activity_id, " +  // 活动ID
                "activity_rule_id, " +  // 活动规则ID
                "order_detail_coupon_id as coupon_id, " +  // 优惠券ID
                "create_time, " +  // 创建时间
                "source_id, " +  // 来源ID
                "source_type_id, " +  // 来源类型ID
                "'' as source_type_name, " +  // 来源类型名称（暂时使用空字符串）
                "split_activity_amount, " +  // 分摊活动金额
                "split_coupon_amount, " +  // 分摊优惠券金额
                "split_total_amount, " +  // 分摊总金额
                "CURRENT_TIMESTAMP() as row_op_ts " +  // 操作时间戳
                "from dwd_order_pre");

        // 创建临时视图便于后续操作
        tableEnv.createTemporaryView("filtered_table", filteredTable);
        // 打印过滤后的数据用于调试
        tableEnv.toAppendStream(filteredTable, Row.class).print(">>>>>>");

        //TODO 4.创建DWD层下单数据表
        // 创建目标Kafka表，用于存储处理后的订单明细数据
        // 表结构与过滤后的数据保持一致
        tableEnv.executeSql("" +
                "create table dwd_trade_order_detail( " +
                "id string, " +  // 订单明细ID
                "order_id string, " +  // 订单ID
                "user_id string, " +  // 用户ID
                "sku_id string, " +  // 商品SKU ID
                "sku_name string, " +  // 商品名称
                "sku_num string, " +  // 商品数量
                "order_price string, " +  // 订单价格
                "province_id string, " +  // 省份ID
                "activity_id string, " +  // 活动ID
                "activity_rule_id string, " +  // 活动规则ID
                "coupon_id string, " +  // 优惠券ID
                "create_time string, " +  // 创建时间
                "source_id string, " +  // 来源ID
                "source_type_id string, " +  // 来源类型ID
                "source_type_name string, " +  // 来源类型名称
                "split_activity_amount string, " +  // 分摊活动金额
                "split_coupon_amount string, " +  // 分摊优惠券金额
                "split_total_amount string, " +  // 分摊总金额
                "row_op_ts timestamp_ltz(3) " +  // 操作时间戳
                ")" + MyKafkaUtil.getKafkaSinkDDL("dwd_trade_order_detail"));

        //TODO 5.将数据写出到Kafka
        // 将过滤处理后的数据写入目标Kafka表
        tableEnv.executeSql("insert into dwd_trade_order_detail select * from filtered_table");
        //TODO 6.启动任务
        // 执行Flink作业
        System.out.println("DwdTradeOrderDetail任务启动中...");
        env.execute("DwdTradeOrderDetail");
    }
}